package dk.almbrand.microservice.kafka;

import dk.almbrand.microservice.circuitbreaker.CircuitBreaker;
import dk.almbrand.microservice.circuitbreaker.CircuitBreakerImpl;
import dk.almbrand.microservice.circuitbreaker.exceptions.ArgumentOutOfRangeException;
import dk.almbrand.microservice.exception.SystemException;
import dk.almbrand.microservice.kafka.consumer.ConsumerDaemon;
import dk.almbrand.microservice.kafka.consumer.ConsumerDaemonImpl;

import dk.almbrand.microservice.kafka.consumer.thread.ConsumerThreadController;
import dk.almbrand.microservice.kafka.consumer.thread.ConsumerThreadControllerImpl;
import dk.almbrand.microservice.kafka.producer.ProducerImpl;
import dk.almbrand.microservice.kafka.producer.Producer;

import dk.almbrand.microservice.kafka.producer.secure.ProtectedProducerImpl;
import dk.almbrand.microservice.kafka.producer.secure.ProtectedProducer;
import dk.almbrand.microservice.service.base.MicroserviceWrapper;


public class KafkaFacadeImpl implements KafkaFacade {
    
    private ConsumerThreadController consumerController;
    private ProtectedProducer protectedProducer;

    public KafkaFacadeImpl(MicroserviceWrapper wrapper, String host, String serializer, String topic) throws ArgumentOutOfRangeException, SystemException {
        Producer producer = new ProducerImpl().setHost(host).setSerializer(serializer).setTopic(topic).build();
        protectedProducer = new ProtectedProducerImpl(producer, new CircuitBreakerImpl(3, 6000 * 1000000));
        
        ConsumerDaemon consumer = new ConsumerDaemonImpl(wrapper).setHost("ubuntu:2181").setGroup("test-group").setTopic("verification-topic");        
        // Start the consumer as a Futuretask and run it in its own thread;
        consumerController = new ConsumerThreadControllerImpl();
        consumerController.start(consumer);
    }
    
    public ProtectedProducer getProducer() {        
        return protectedProducer;
    }

    public ConsumerThreadController getConsumerThread() {
        return consumerController;
    }
}
